package org.apache.flink.runtime.util;

import org.apache.flink.configuration.ClusterOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.slotpool.LocationPreferenceSlotSelectionStrategy;
import org.apache.flink.runtime.slotpool.PreviousAllocationSlotSelectionStrategy;
import org.apache.flink.runtime.slotpool.SlotSelectionStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Utility class for selecting {@link SlotSelectionStrategy}. */
public class SlotSelectionStrategyUtils {

    private static final Logger LOG = LoggerFactory.getLogger(SlotSelectionStrategyUtils.class);

    public static SlotSelectionStrategy selectSlotSelectionStrategy(final Configuration configuration) {
        final boolean evenlySpreadOutSlots =
                configuration.getBoolean(ClusterOptions.EVENLY_SPREAD_OUT_SLOTS_STRATEGY);

        final SlotSelectionStrategy locationPreferenceSlotSelectionStrategy;

        locationPreferenceSlotSelectionStrategy =
                evenlySpreadOutSlots
                        ? LocationPreferenceSlotSelectionStrategy.createEvenlySpreadOut()
                        : LocationPreferenceSlotSelectionStrategy.createDefault();
        
        return PreviousAllocationSlotSelectionStrategy.create(
                locationPreferenceSlotSelectionStrategy);
        
        // return locationPreferenceSlotSelectionStrategy;

    }

    /** Private default constructor to avoid being instantiated. */
    private SlotSelectionStrategyUtils() {}
}
